package org.zhadoop.stream.cql.tasks;

import java.util.Collections;
import java.util.List;

import org.abego.treelayout.internal.util.java.lang.string.StringUtil;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.stream.api.Application;
import org.zhadoop.stream.api.streams.Schema;
import org.zhadoop.stream.cql.CQLResult;
import org.zhadoop.stream.cql.DriverContext;
import org.zhadoop.stream.cql.builder.ApplicationBuilder;
import org.zhadoop.stream.cql.exception.CQLException;
import org.zhadoop.stream.cql.exception.ExecutorException;
import org.zhadoop.stream.cql.exception.SemanticAnalyzerException;
import org.zhadoop.stream.cql.hooks.SemanticAnalyzeHook;
import org.zhadoop.stream.cql.semanticanalyzer.SemanticAnalyzer;
import org.zhadoop.stream.cql.semanticanalyzer.SemanticAnalyzerFactory;
import org.zhadoop.stream.cql.semanticanalyzer.analyzecontext.AnalyzeContext;
import org.zhadoop.stream.cql.semanticanalyzer.analyzecontext.SubmitApplicationAnalyzeContext;
import org.zhadoop.stream.cql.semanticanalyzer.parser.context.ParseContext;
import org.zhadoop.streaming.config.StreamingConfig;
import org.zhadoop.streaming.exception.ErrorCode;
import org.zhadoop.streaming.exception.StreamingRuntimeException;

import com.google.common.collect.Lists;

/**
 * 提交应用程序执行task
 * @author zwd
 *
 */
public class SubmitTask extends BaseTask {

	private static final Logger LOG = LoggerFactory.getLogger(SubmitTask.class);
	
	private DriverContext context;
	
	private List<AnalyzeContext> analyzeContexts;
	
	private SubmitApplicationAnalyzeContext submitContext;
	
	private String[] resultHeader = {"result"};
	
	private String result = null;
	
	private List<Schema> emptySchemas = Collections.emptyList();
	
	/**
	 * 超过会自动扩展
	 */
	private String format = "%-20s";
	
	
	
	@Override
	public void init(DriverContext driverContext, StreamingConfig config, List<SemanticAnalyzeHook> analyzeHooks)
			throws CQLException {
		super.init(driverContext, config, analyzeHooks);
		context = driverContext;
		analyzeContexts = Lists.newArrayList();
	}

	@Override
	public void execute(ParseContext parseContext) throws CQLException {
		if(parseContext == null){
			LOG.error("ParseContext is null.");
			throw new CQLException(ErrorCode.SEMANTICANALYZE_CONTEXT_NULL);
		}
		
		 securityPrepare();
		 parseSubmit(parseContext);
		 createApplication();
		 dropApplicationIfAllow();
		 submitApplication();

	}


	/**
	 * 安全准备
	 */
	private void securityPrepare() {
		// TODO Auto-generated method stub
		
	}
	/**
	 * 解析前的工作
	 * @param parseContext
	 * @throws SemanticAnalyzerException
	 */
	private void parseSubmit(ParseContext parseContext) throws SemanticAnalyzerException{
		SemanticAnalyzer analyzer = SemanticAnalyzerFactory.createAnalyzer(parseContext, emptySchemas);
		submitContext = (SubmitApplicationAnalyzeContext) analyzer.analyze();
	}
	
	private void submitApplication() {
		// TODO Auto-generated method stub
		
	}

	private void dropApplicationIfAllow() {
		// TODO Auto-generated method stub
		
	}

	

	private void createApplication() throws CQLException{
		String savePath = submitContext.getPhysicPlanPath();
		String appName = submitContext.getAppName();
		
		if(!StringUtils.isEmpty(savePath)){
			
		}
		
		context.setApp(createAPIApplication(appName));
	}
	
	
	private Application createAPIApplication(String appName) throws CQLException{
		//检查application的名称是否为空
		checkApplicationNameIsEmpty(appName);
		Application app = null;
		if(context.getApp() == null){
			semanticAnalyzerLazyContexts();
			app = new ApplicationBuilder().build(appName, analyzeContexts, context);
		}else{
			app = context.getApp();
		}
		app.setApplicationId(appName);
		return app;
	}

	private void checkApplicationNameIsEmpty(String appName) throws ExecutorException{
		if(StringUtils.isEmpty(appName)){
			ExecutorException exception = new ExecutorException(ErrorCode.TOP_NO_NAME);
			LOG.error("Application name is null.",exception);
			throw exception;
		}
	}
	
	private void semanticAnalyzerLazyContexts() throws SemanticAnalyzerException{
		try{
			for(ParseContext parseContext : context.getParseContexts()){
				preAnalyze(context,parseContext);
				SemanticAnalyzer analyzer = SemanticAnalyzerFactory.createAnalyzer(parseContext, context.getSchemas());
				AnalyzeContext analyzeContext = analyzer.analyze();
				postAnalyze(context, analyzeContext, parseContext);
				analyzeContexts.add(analyzeContext);
			}
		}catch (StreamingRuntimeException e) {
			//防止表达式创建或者子实例创建的时候抛出runtime异常
			//不过这里只捕获StreamingRuntimeException异常
			//RuntimeException计划后期全部替换为StreamingRuntimeException
			throw SemanticAnalyzerException.wrapStreamingRunTimeException(e);
		}
	}

	@Override
	public CQLResult getResult() {
		// TODO Auto-generated method stub
		return null;
	}

}
